#include "gvj_server.h"

#define SERVER_ADDR 		INADDR_ANY
#define SERVER_PORT 		7726
#define LISTENQ 			5
#define MAX_EPOLL_EVENTS 	100

#define FUNC_OK				0
#define SOCKET_CLOSE		1
#define IN_PIPE_CLOSE		2
#define RECV_ERROR			3
#define SEND_ERROR			4

const key_t GvjServer::MSG_KEY 			= 0x7726;
const int GvjServer::MSG_CREAT_FLAG 	= IPC_CREAT|666;
const int GvjServer::MSG_TEST_FLAG 		= IPC_CREAT|IPC_EXCL|666;
const int GvjServer::MSG_EXCL_FLAG 		= IPC_EXCL;
const int GvjServer::HEADER_SIZE		= 64;
const int GvjServer::FIELD_SIZE			= 16;

int GvjServer::main_process(std::vector<std::string> oj) {
	// 修改进程名称
	set_process_name("gvj_server");

	// 初始化map_pipe(创建与代理进程通信的管道)，并fork出代理进程
	for (int i = 0; i < oj.size(); i++) {
		create_proxy_process(oj[i].data(), "/root/project/GVJ/cookie/");
	}
	/*
	for (int i = 0; i < oj.size(); i++) {
		int *p1 = map_pipe[oj[i]].first;
		int *p2 = map_pipe[oj[i]].second;
		close(p1[0]);
		close(p2[1]);
	}
	*/

	// 初始化map_oj_obj
	for (int i = 0; i < oj.size(); i++) {
		if (oj[i] == "poj") 
			map_oj_obj[oj[i]] = new Poj();
		else if (oj[i] == "hdu") 
			map_oj_obj[oj[i]] = new Hdu();
	}
	// 初始化map_que
	for (int i = 0; i < oj.size(); i++) {
		map_que[oj[i]] = new std::queue<PipeSendBuff *>();
	}
	
	// 初始化map_register
	for (int i = 0; i < oj.size(); i++) {
		map_register[oj[i]] = false;
	}

	// 创建消息队列，消息队列可能已经存在，清空消息队列
	msg_id = msgget(MSG_KEY, MSG_TEST_FLAG);
	if (msg_id == -1) {
		msg_id = msgget(MSG_KEY, MSG_EXCL_FLAG);
		msgctl(msg_id, IPC_RMID, NULL);
		msg_id = msgget(MSG_KEY, MSG_CREAT_FLAG);
	}

	// fork出日志进程

	// 测试与代理进程之间的管道
	/*
	std::map<std::string, std::pair<int*, int*> >::iterator it;
	for (it = map_pipe.begin(); it != map_pipe.end(); it++) {
		int in_pipe = it->second.second[0];
		char buff[100];
		ssize_t n;
		memset(buff, 0, sizeof(buff));
		n = read(in_pipe, buff, 100);
		printf("%s\n", buff);
	}
	for (it = map_pipe.begin(); it != map_pipe.end(); it++) {
		int out_pipe = it->second.first[1];
		char buff[100];
		ssize_t n;
		memset(buff, 0, sizeof(buff));
		sprintf(buff, "hello i am listen");
		n = write(out_pipe, buff, strlen(buff));
		printf("listen write:%d to %s\n", n, it->first.data());
	}
	*/
	// 测试与代理进程之间的管道
	
	// DEBUG
	//while (true) {
	//for (int i = 0; i < 10; i++) {
	//std::string stream = test_create_stream_to_proxy("poj");
	//int out_pipe = map_pipe["poj"].first[1];
	//printf("%s\n", stream.data());
	//ssize_t nbytes = write(out_pipe, stream.data(), strlen(stream.data()));
	//printf("nbytes = %d\n", nbytes);
	//}
	//sleep(120);
	//}
	//// DEBUG
	//exit(0);
	// step 6: 进入监听流程
	listen_process();
}

int GvjServer::listen_process() {
	// 设置服务器地址结构;创建监听socket;设置监听socket非阻塞;bind和listen
	set_sockaddr_in(&server_addr);
	listen_fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	set_non_blocking(listen_fd);
	bind(listen_fd, (const struct sockaddr*)&server_addr, sizeof(server_addr));
	listen(listen_fd, LISTENQ);
	// 设置服务器地址结构;创建监听socket;设置监听socket非阻塞;bind和listen

	// 创建保存就绪事件的结构体数组;创建epoll句柄
	ep_fd = epoll_create(MAX_EPOLL_EVENTS);
	ready_evs = new epoll_event[MAX_EPOLL_EVENTS];
	// 创建保存就绪事件的结构体数组;创建epoll句柄

	// 将监听socket加入epoll
	ev.events = EPOLLIN;
	ev.data.ptr    = new EpollData(LISTEN_SOCKET_TYPE, listen_fd, NULL);
	epoll_ctl(ep_fd, EPOLL_CTL_ADD, listen_fd, &ev);
	// 将监听socket加入epoll

	// 将管道的读端加入epoll
	std::map<std::string, std::pair<int*, int*> >::iterator it;
	for (it = map_pipe.begin(); it != map_pipe.end(); it++) {
		int in_pipe = it->second.second[0];
		ev.events = EPOLLIN;
		PipeRecvBuff *prb = new PipeRecvBuff();
		ev.data.ptr  = new EpollData(PIPE_RECV_TYPE, in_pipe, (void*)prb);
		epoll_ctl(ep_fd, EPOLL_CTL_ADD, in_pipe, &ev);
	}
	// 将管道的读端加入epoll

	while (true) {
		int nfds = epoll_wait(ep_fd, ready_evs, MAX_EPOLL_EVENTS, 120);
		//printf("epoll_wait return\n");
		for (int i = 0; i < nfds; i++) {
			EpollData *ep_data = (EpollData *)ready_evs[i].data.ptr;
			if (ep_data->type == LISTEN_SOCKET_TYPE) {
				int res = listen_socket_ready();
			}
			else if (ep_data->type == SOCKET_RECV_TYPE) {
				int res = client_socket_recv_ready(ep_data);
			}
			else if (ep_data->type == PIPE_SEND_TYPE) {
				int res = pipe_send_ready(ep_data);
			}
			else if (ep_data->type == PIPE_RECV_TYPE) {
				int res = pipe_recv_ready(ep_data);
			}
			else if (ep_data->type == SOCKET_SEND_TYPE) {
				int res = client_socket_send_ready(ep_data);
			}
		}
		check_map_list();
		
		// DEBUG
		//while (true) {
		//for (int i = 0; i < 3; i++) {
		//std::string stream = test_create_stream_to_proxy("poj");
		//int out_pipe = map_pipe["poj"].first[1];
		//printf("%s\n", stream.data());
		//ssize_t nbytes = write(out_pipe, stream.data(), strlen(stream.data()));
		//printf("nbytes = %d\n", nbytes);
		//}
		//sleep(30);
		//}
		//// DEBUG
	}

	return 0;
}

int GvjServer::listen_socket_ready() {
	// 每次最多接入50个客户，剩下的客户到下一次accept
	int max_accept_client = 50;
	printf("listen_socket_ready\n");
	while (max_accept_client--) {
		memset(&client_addr, 0, sizeof(client_addr));
		client_addr_len = sizeof(struct sockaddr_in);

		client_fd = accept(listen_fd, (struct sockaddr *)&client_addr, &client_addr_len);
		printf("(%d)client_fd: %d\n", max_accept_client, client_fd);
		if (client_fd == -1) {
			printf("errno: %d\n", errno);
			printf("%s\n", strerror(errno));
			break;
		}

		// 为client_fd设置一个struct epoll_event，注册到epoll中
		ev.events = EPOLLIN;
		SocketRecvBuff *srb = new SocketRecvBuff();
		EpollData *ep_data 	= new EpollData(SOCKET_RECV_TYPE, client_fd, (void*)srb);
		ev.data.ptr = ep_data;
		epoll_ctl(ep_fd, EPOLL_CTL_ADD, client_fd, &ev);
	}
}

int GvjServer::client_socket_recv_ready(EpollData *ep_data) {
    int fd = ep_data->fd;
    SocketRecvBuff *srb = (SocketRecvBuff *)ep_data->data;
	
	// 连header都没有完全接收
    if (!srb->finish_header()) {
		printf("还没完全接收客户端的请求头\n");
		ssize_t n = srb->recv_header(fd);
		if (n == -1) {
			return RECV_ERROR;
		}
		if (n == 0) {
			close_socket(fd, ep_data, srb);
			return SOCKET_CLOSE;
		}
		if (srb->finish_header()) {
			printf("完全接收客户端请求头，开始过滤请求头\n");
			if (request_filter(srb->header, map_oj_obj)) {
				printf("客户端请求头非法，主动关闭TCP连接\n");
				close_socket(fd, ep_data, srb);
				return SOCKET_CLOSE;
			}
			printf("客户端请求头合法，提取OJ名，代码长度，接收body\n");
			//write(1, srb->header, 64), printf("\n");
			srb->get_oj();
			srb->get_body_length();
			//printf("body长度：%d\n", srb->body_length);
		}
    }

	// 已经完全接收header，进程过滤，然后接收body
    if (srb->finish_header() && !srb->finish_body()) {
		printf("接收客户端body\n");
		ssize_t n = srb->recv_body(fd);
		if (n == -1) {
			return RECV_ERROR;
		}
		else if (n == 0) {
			close_socket(fd, ep_data, srb);
			return SOCKET_CLOSE;
		}
    }

    // 已经完全接收
    if (srb->finish_body()) {
		printf("完全接收客户端的请求包:\n");
		write(1, srb->header, 64);
		write(1, srb->body, srb->body_length), printf("\n");
		std::string stream = serialize(fd, srb->header, srb->body);
		printf("监听进程序列后的流：%s\n", stream.data());
		std::queue<PipeSendBuff*> *que = map_que[srb->oj];
		PipeSendBuff *psb = new PipeSendBuff(stream.data());
		que->push(psb);

		delete ep_data;
		delete srb;
		epoll_ctl(ep_fd, EPOLL_CTL_DEL, fd, NULL);
    }

    return FUNC_OK;
}

// 监听进程有评判任务给代理进程处理
int GvjServer::pipe_send_ready(EpollData *ep_data) {	  
	printf("pipe_send_ready函数: 开始写\n");
	int fd = ep_data->fd;
	PipeSendBuff *psb = (PipeSendBuff *)ep_data->data;
	if (!psb->finish()) {
		ssize_t n = psb->send_data(fd);
		// 可能是write缓冲区不够，errno是EAGIN
		// 也可能是代理进程已经挂了，产生了SIGPIPE
		printf("监听进程写管道:n=%d\n", n);
		if (n == -1) {
			if (errno == EPIPE) {
				// 代理进程已经挂了，重启代理进程
				// 将pb的内容重新发送
				//pb->resend();
			}
			return SEND_ERROR;
		}
	}
	
	if (psb->finish()) {
		printf("out_pipe的写入操作完成，将out_pipe从epoll中删除\n");
		epoll_ctl(ep_fd, EPOLL_CTL_DEL, fd, NULL);
		std::map<std::string, std::pair<int*, int*> >::iterator it;
		for (it = map_pipe.begin(); it != map_pipe.end(); it++) {
			int out_pipe = it->second.first[1];
			if (out_pipe == fd) break;
		}
		map_register[it->first] = false;
		delete psb;
		delete ep_data;
	}
}

int GvjServer::pipe_recv_ready(EpollData *ep_data) {
	int fd = ep_data->fd;
	PipeRecvBuff *prb = (PipeRecvBuff *)ep_data->data;
	// 还没接收长度
	if (!prb->finish_length()) {
		//printf("监听进程从管道读评判结果，还没读到长度\n");
		ssize_t n = prb->recv_length(fd);
		if (n == -1) return RECV_ERROR;
		else if (n == 0) return IN_PIPE_CLOSE;
		
		if (prb->finish_length()) {
			printf("监听进程从管道读评判结果，读完长度\n");
			prb->get_length();
		}
	}
	
	// 只接收了长度
	if (prb->finish_length() && !prb->finish_data()) {
		ssize_t n = prb->recv_data(fd);
		if (n == -1) return RECV_ERROR;
		else if (n == 0) return IN_PIPE_CLOSE;
	}

	// 完全接收
	if (prb->finish_data()) {
		printf("监听进程从管道中完整读取了评判结果:\n");
		printf("%s %s\n", prb->length_buff, prb->data_buff);
		EpollData *new_ep_data = deserialize(prb->data_buff);
		printf("客户端socket fd: %d\n", new_ep_data->fd);
		SocketSendBuff *ssb = (SocketSendBuff *)new_ep_data->data;
		write(1, ssb->body, ssb->body_length), printf("\n");
		ev.events = EPOLLOUT;
		ev.data.ptr = new_ep_data;
		epoll_ctl(ep_fd, EPOLL_CTL_ADD, new_ep_data->fd, &ev);
		prb->reset();
	}

	return FUNC_OK;
}

int GvjServer::client_socket_send_ready(EpollData *ep_data) {
	int fd = ep_data->fd;
	SocketSendBuff *ssb = (SocketSendBuff *)ep_data->data;
	
	if (!ssb->finish_header()) {
		ssize_t n = ssb->send_header(fd);
		// 可能对端socket已经关闭，或其他信号中断或错误
		if (n == -1) {
			if (errno == EPIPE) {
				close_socket(fd, ep_data, ssb);
				return SOCKET_CLOSE;
			}
			return SEND_ERROR;
		}
		if (ssb->finish_header()) {
			ssb->get_body_length();
		}
	}

	if (ssb->finish_header() && !ssb->finish_body()) {
		ssize_t n = ssb->send_body(fd);
		if (n == -1) {
			if (errno == EPIPE) {
				close_socket(fd, ep_data, ssb);
				return SOCKET_CLOSE;
			}
			return SEND_ERROR;
		}
	}

	// 发送完成
	if (ssb->finish_body()) {
		printf("监听进程终于将json发送给客户端了\n");
		delete ep_data;
		delete ssb;
		ev.events = EPOLLIN;
		SocketRecvBuff *new_srb = new SocketRecvBuff();
		EpollData  *new_ep_data = new EpollData(SOCKET_RECV_TYPE, fd, new_srb);
		ev.data.ptr = new_ep_data;
		epoll_ctl(ep_fd, EPOLL_CTL_MOD, fd, &ev);
	}
}

// 检查每个out_pipe是否有发送列表
// 根据发送列表是否为空，修改epoll是否监控out_pipe
void GvjServer::check_map_list() {
	std::map<std::string, std::queue<PipeSendBuff*>* >::iterator it;
	for (it = map_que.begin(); it != map_que.end(); it++) {
		std::string oj = it->first;
		int out_pipe = map_pipe[oj].first[1]; 
		std::queue<PipeSendBuff*> *que = it->second;
		bool excl = map_register[oj];
		if (!excl && !que->empty()) {
			printf("通向%s的管道(%d)可写，向epoll注册\n", it->first.data(), out_pipe);
			PipeSendBuff *psb = que->front();
			que->pop();
			EpollData *ep_data = new EpollData(PIPE_SEND_TYPE, out_pipe, psb);
			ev.events = EPOLLOUT;
			ev.data.ptr = ep_data;
			epoll_ctl(ep_fd, EPOLL_CTL_ADD, out_pipe, &ev);
			map_register[oj] = true;
		}
	}
}



void GvjServer::create_proxy_process(const char *oj, const char *cookie_dir) {
	int *pipe_ptr1 = new int[2];
	int *pipe_ptr2 = new int[2];

	pipe(pipe_ptr1);
	pipe(pipe_ptr2);

	set_non_blocking(pipe_ptr1[0]);
	set_non_blocking(pipe_ptr1[1]);
	set_non_blocking(pipe_ptr2[0]);
	set_non_blocking(pipe_ptr2[1]);

	map_pipe[oj] = std::make_pair(pipe_ptr1, pipe_ptr2);
		
	proxy_starter(map_pipe, oj, msg_id, cookie_dir);
	close(pipe_ptr1[0]);
	close(pipe_ptr2[1]);
}

void GvjServer::set_sockaddr_in(struct sockaddr_in *sa) {
	memset(sa, 0, sizeof(struct sockaddr_in));
	sa->sin_family = AF_INET;
	sa->sin_addr.s_addr = SERVER_ADDR;
	sa->sin_port = htons(SERVER_PORT);
}

int GvjServer::close_socket(int fd, EpollData *ep_data, Buff *buff) {
	printf("close_socket (%d)\n", fd);
	epoll_ctl(ep_fd, EPOLL_CTL_DEL, fd, NULL);
	delete ep_data;
	delete buff;
	return close(fd);
}


